package com.pengpeng.rocketmq.web;

import com.pengpeng.rocketmq.consumer.Consumer;
import com.pengpeng.rocketmq.producer.Producer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CountDownLatch;

/**
 * @author 彭鹏
 * @date 2021/10/20.
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class Controller {

    @Autowired
    private Producer producer;

    /**
     * 可靠的同步传输广泛应用于大量场景，如重要通知消息、短信通知、短信营销系统等。
     * @return
     * @throws Exception
     */
    @RequestMapping("/syncProducer")
    public Object syncProducer() throws Exception {
        //总共发送五次消息
        for (int i = 0; i < 20; i++) {
            //创建生产信息
            Message message = new Message(Consumer.TOPIC, "TagA", ("Hello RocketMQ:" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //发送
            SendResult sendResult = producer.getProducer().send(message);
            log.info("输出生产者信息={}",sendResult);
        }
        return "成功";
    }

    /**
     * 异步传输通常用于响应时间敏感型业务场景。
     * @return
     * @throws Exception
     */
    @RequestMapping("/asyncProducer")
    public Object callback() throws Exception {
        //总共发送五次消息
        int messageCount = 20;
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0;i<messageCount;i++) {
            try {

                final int index = i;
                //创建生产信息
                Message message = new Message(Consumer.TOPIC, "TagA", "OrderID188", ("Hello World").getBytes(RemotingHelper.DEFAULT_CHARSET));
                //发送
                producer.getProducer().send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return "成功";
    }

    /**
     * 单向传输用于需要中等可靠性的情况，例如日志收集。
     * @return
     * @throws Exception
     */
    @RequestMapping("/onewayProducer")
    public Object onewayProducer() throws Exception {
        //总共发送五次消息
        for (int i = 0; i < 20; i++) {
            //创建生产信息
            Message message = new Message(Consumer.TOPIC, "TagA", ("Hello RocketMQ:" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //发送
            producer.getProducer().sendOneway(message);
        }
        Thread.sleep(5000);
        return "成功";
    }
}
